#pragma once

#include <boost/asio.hpp>
#include <boost/asio/posix/stream_descriptor.hpp>
#include <sys/eventfd.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <fcntl.h>
#include <unistd.h>
#include <string>
#include <vector>
#include <unordered_map>
#include <memory>
#include <functional>
#include <atomic>
#include <thread>
#include <mutex>
#include <set>

// 包含现有的msgq模块
#include "../include/msgq/msgq.h"

// 包含调试工具
#include "debug_utils.h"

namespace improved_pubsub {

// 消息结构
struct Message {
    std::string topic;
    std::vector<uint8_t> data;
    uint64_t timestamp;
    uint64_t sequence_id;
    
    Message() = default;
    Message(const std::string& t, const std::vector<uint8_t>& d) 
        : topic(t), data(d), timestamp(0), sequence_id(0) {}
};

// 跨进程通信配置
struct Config {
    std::string registry_socket_path = "/tmp/pubsub_registry";
    std::string msgq_prefix = "pubsub_";
    size_t topic_buffer_size = DEFAULT_SEGMENT_SIZE;  // 使用msgq的默认大小
    int max_subscribers_per_topic = 64;
};

// 消息回调函数类型
using MessageCallback = std::function<void(const Message&)>;

// 改进的EventFD工具类
class EventFdHelper {
public:
    // 通过Unix Socket发送文件描述符
    static int send_fd(int socket_fd, int fd_to_send) {
        DEMO_DEBUG("EventFdHelper::send_fd: 准备通过socket {} 发送fd {}", socket_fd, fd_to_send);
        
        struct msghdr msg;
        memset(&msg, 0, sizeof(msg));
        char buf[CMSG_SPACE(sizeof(fd_to_send))];
        memset(buf, '\0', sizeof(buf));
        
        struct iovec io = { .iov_base = (void*)"FD", .iov_len = 2 };
        msg.msg_iov = &io;
        msg.msg_iovlen = 1;
        
        msg.msg_control = buf;
        msg.msg_controllen = sizeof(buf);
        
        struct cmsghdr* cmsg = CMSG_FIRSTHDR(&msg);
        cmsg->cmsg_level = SOL_SOCKET;
        cmsg->cmsg_type = SCM_RIGHTS;
        cmsg->cmsg_len = CMSG_LEN(sizeof(fd_to_send));
        
        *((int*)CMSG_DATA(cmsg)) = fd_to_send;
        
        int result = sendmsg(socket_fd, &msg, 0);
        if (result > 0) {
            DEMO_INFO("EventFdHelper::send_fd: 成功发送fd {} 通过socket {}", fd_to_send, socket_fd);
        } else {
            DEMO_ERROR("EventFdHelper::send_fd: 发送fd失败, socket={}, fd={}, errno={}", 
                       socket_fd, fd_to_send, errno);
        }
        return result;
    }
    
    // 通过Unix Socket接收文件描述符
    static int recv_fd(int socket_fd) {
        DEMO_DEBUG("EventFdHelper::recv_fd: 准备从socket {} 接收fd", socket_fd);
        
        struct msghdr msg = {0};
        char m_buffer[256];
        char c_buffer[256];
        
        struct iovec io = { .iov_base = m_buffer, .iov_len = sizeof(m_buffer) };
        msg.msg_iov = &io;
        msg.msg_iovlen = 1;
        msg.msg_control = c_buffer;
        msg.msg_controllen = sizeof(c_buffer);
        
        if (recvmsg(socket_fd, &msg, 0) < 0) {
            DEMO_ERROR("EventFdHelper::recv_fd: recvmsg失败, socket={}, errno={}", socket_fd, errno);
            return -1;
        }
        
        struct cmsghdr* cmsg = CMSG_FIRSTHDR(&msg);
        if (cmsg && cmsg->cmsg_level == SOL_SOCKET && cmsg->cmsg_type == SCM_RIGHTS) {
            int received_fd = *((int*)CMSG_DATA(cmsg));
            DEMO_INFO("EventFdHelper::recv_fd: 成功接收fd {} 从socket {}", received_fd, socket_fd);
            return received_fd;
        }
        
        DEMO_ERROR("EventFdHelper::recv_fd: 未收到有效的文件描述符, socket={}", socket_fd);
        return -1;
    }
};

// Topic信息（改进版本，基于msgq）
struct TopicInfo {
    std::string topic_name;
    msgq_queue_t msgq;              // 使用现有的msgq作为消息队列
    std::atomic<uint64_t> message_count;
    std::set<int> subscriber_eventfds;  // 所有订阅者的EventFD列表
    std::mutex eventfds_mutex;
    bool is_initialized;
    
    TopicInfo() : message_count(0), is_initialized(false) {}
    
    ~TopicInfo() {
        if (is_initialized) {
            msgq_close_queue(&msgq);
            // 清理所有EventFD
            std::lock_guard<std::mutex> lock(eventfds_mutex);
            for (int fd : subscriber_eventfds) {
                close(fd);
            }
        }
    }
    
    bool initialize(const std::string& name, size_t buffer_size) {
        DEMO_DEBUG("TopicInfo::initialize: 初始化topic '{}', buffer_size={}", name, buffer_size);
        
        topic_name = name;
        std::string msgq_path = "pubsub_" + name;
        
        if (msgq_new_queue(&msgq, msgq_path.c_str(), buffer_size) < 0) {
            DEMO_ERROR("TopicInfo::initialize: msgq_new_queue失败, topic={}, path={}", name, msgq_path);
            return false;
        }
        
        msgq_init_publisher(&msgq);
        is_initialized = true;
        DEMO_INFO("TopicInfo::initialize: 成功初始化topic '{}', msgq_path={}", name, msgq_path);
        return true;
    }
    
    void add_subscriber_eventfd(int eventfd) {
        std::lock_guard<std::mutex> lock(eventfds_mutex);
        subscriber_eventfds.insert(eventfd);
        DEMO_INFO("TopicInfo::add_subscriber_eventfd: topic '{}' 添加订阅者eventfd={}, 总订阅者数={}", 
                  topic_name, eventfd, subscriber_eventfds.size());
    }
    
    void remove_subscriber_eventfd(int eventfd) {
        std::lock_guard<std::mutex> lock(eventfds_mutex);
        subscriber_eventfds.erase(eventfd);
        close(eventfd);
        DEMO_INFO("TopicInfo::remove_subscriber_eventfd: topic '{}' 移除订阅者eventfd={}, 剩余订阅者数={}", 
                  topic_name, eventfd, subscriber_eventfds.size());
    }
    
    void notify_all_subscribers() {
        std::lock_guard<std::mutex> lock(eventfds_mutex);
        uint64_t notification = 1;
        int success_count = 0;
        int fail_count = 0;
        
        DEMO_TRACE("TopicInfo::notify_all_subscribers: topic '{}' 通知 {} 个订阅者", 
                   topic_name, subscriber_eventfds.size());
        
        for (int eventfd : subscriber_eventfds) {
            // 忽略写入失败（订阅者可能已经退出）
            if (write(eventfd, &notification, sizeof(notification)) == sizeof(notification)) {
                success_count++;
                DEMO_TRACE("TopicInfo::notify_all_subscribers: 成功通知eventfd={}", eventfd);
            } else {
                fail_count++;
                DEMO_WARN("TopicInfo::notify_all_subscribers: 通知eventfd={}失败, errno={}", eventfd, errno);
            }
        }
        
        if (subscriber_eventfds.size() > 0) {
            DEMO_DEBUG("TopicInfo::notify_all_subscribers: topic '{}' 通知完成, 成功={}, 失败={}", 
                       topic_name, success_count, fail_count);
        }
    }
};

// 客户端请求协议
enum class RequestType : uint8_t {
    PUBLISH_CONNECT = 1,     // Publisher连接到topic
    SUBSCRIBE_CONNECT = 2,   // Subscriber连接到topic
    NOTIFY_NEW_SUBSCRIBER = 3 // 通知Publisher有新订阅者（从注册中心到Publisher）
};

struct ClientRequest {
    RequestType type;
    char topic_name[64];
};

// Topic注册服务（改进版本）
class TopicRegistry {
private:
    Config config_;
    int server_socket_;
    std::unordered_map<std::string, std::shared_ptr<TopicInfo>> topics_;
    std::mutex topics_mutex_;
    boost::asio::io_context io_context_;
    std::unique_ptr<std::thread> service_thread_;
    std::atomic<bool> running_;
    
public:
    explicit TopicRegistry(const Config& config) 
        : config_(config), server_socket_(-1), running_(false) {}
    
    ~TopicRegistry() {
        stop();
    }
    
    bool start() {
        DEMO_INFO("TopicRegistry::start: 启动注册中心服务, socket_path={}", config_.registry_socket_path);
        
        // 创建Unix Domain Socket服务器
        server_socket_ = socket(AF_UNIX, SOCK_STREAM, 0);
        if (server_socket_ == -1) {
            DEMO_ERROR("TopicRegistry::start: 创建socket失败, errno={}", errno);
            return false;
        }
        DEMO_DEBUG("TopicRegistry::start: 创建socket成功, fd={}", server_socket_);
        
        struct sockaddr_un addr;
        memset(&addr, 0, sizeof(addr));
        addr.sun_family = AF_UNIX;
        strncpy(addr.sun_path, config_.registry_socket_path.c_str(), 
                sizeof(addr.sun_path) - 1);
        
        // 删除已存在的socket文件
        unlink(config_.registry_socket_path.c_str());
        DEMO_DEBUG("TopicRegistry::start: 清理旧socket文件: {}", config_.registry_socket_path);
        
        if (bind(server_socket_, (struct sockaddr*)&addr, sizeof(addr)) == -1) {
            DEMO_ERROR("TopicRegistry::start: bind失败, path={}, errno={}", config_.registry_socket_path, errno);
            close(server_socket_);
            return false;
        }
        DEMO_DEBUG("TopicRegistry::start: bind成功, path={}", config_.registry_socket_path);
        
        if (listen(server_socket_, 10) == -1) {
            DEMO_ERROR("TopicRegistry::start: listen失败, errno={}", errno);
            close(server_socket_);
            return false;
        }
        DEMO_DEBUG("TopicRegistry::start: listen成功, 等待连接...");
        
        running_ = true;
        service_thread_ = std::make_unique<std::thread>([this]() {
            accept_loop();
        });
        
        DEMO_INFO("TopicRegistry::start: 注册中心启动成功");
        return true;
    }
    
    void stop() {
        DEMO_INFO("TopicRegistry::stop: 正在停止注册中心服务");
        running_ = false;
        
        if (server_socket_ != -1) {
            DEMO_DEBUG("TopicRegistry::stop: 关闭服务器socket {}", server_socket_);
            close(server_socket_);
            server_socket_ = -1;
        }
        
        if (service_thread_ && service_thread_->joinable()) {
            DEMO_DEBUG("TopicRegistry::stop: 等待服务线程结束");
            service_thread_->join();
        }
        
        // 清理topic
        cleanup_topics();
        DEMO_INFO("TopicRegistry::stop: 注册中心服务已停止");
    }
    
    std::shared_ptr<TopicInfo> get_or_create_topic(const std::string& topic_name) {
        std::lock_guard<std::mutex> lock(topics_mutex_);
        
        auto it = topics_.find(topic_name);
        if (it != topics_.end()) {
            DEMO_DEBUG("TopicRegistry::get_or_create_topic: 找到已存在topic '{}'", topic_name);
            return it->second;
        }
        
        DEMO_INFO("TopicRegistry::get_or_create_topic: 创建新topic '{}', buffer_size={}", 
                  topic_name, config_.topic_buffer_size);
        
        // 创建新topic
        auto topic_info = std::make_shared<TopicInfo>();
        if (!topic_info->initialize(topic_name, config_.topic_buffer_size)) {
            DEMO_ERROR("TopicRegistry::get_or_create_topic: 初始化topic '{}'失败", topic_name);
            return nullptr;
        }
        
        topics_[topic_name] = topic_info;
        DEMO_INFO("TopicRegistry::get_or_create_topic: 成功创建topic '{}', 当前topic总数={}", 
                  topic_name, topics_.size());
        return topic_info;
    }
    
private:
    // 向Publisher发送已存在的订阅者EventFD列表
    void send_existing_subscriber_eventfds(int client_socket, std::shared_ptr<TopicInfo> topic_info, const std::string& topic_name) {
        DEMO_DEBUG("TopicRegistry::send_existing_subscriber_eventfds: 向Publisher发送topic '{}'的已存在订阅者列表", topic_name);
        
        std::lock_guard<std::mutex> lock(topic_info->eventfds_mutex);
        uint32_t subscriber_count = topic_info->subscriber_eventfds.size();
        
        // 发送订阅者数量
        if (send(client_socket, &subscriber_count, sizeof(subscriber_count), 0) != sizeof(subscriber_count)) {
            DEMO_ERROR("TopicRegistry::send_existing_subscriber_eventfds: 发送订阅者数量失败, topic='{}', count={}", 
                       topic_name, subscriber_count);
            return;
        }
        
        DEMO_DEBUG("TopicRegistry::send_existing_subscriber_eventfds: 发送订阅者数量 {} for topic '{}'", 
                   subscriber_count, topic_name);
        
        // 发送每个EventFD
        int success_count = 0;
        for (int eventfd : topic_info->subscriber_eventfds) {
            // 复制EventFD以便发送给Publisher
            int dup_eventfd = dup(eventfd);
            if (dup_eventfd == -1) {
                DEMO_WARN("TopicRegistry::send_existing_subscriber_eventfds: dup eventfd失败, original={}, errno={}", 
                          eventfd, errno);
                continue;
            }
            
            if (EventFdHelper::send_fd(client_socket, dup_eventfd) > 0) {
                success_count++;
                DEMO_DEBUG("TopicRegistry::send_existing_subscriber_eventfds: 成功发送eventfd={} (dup of {})", 
                           dup_eventfd, eventfd);
            } else {
                DEMO_WARN("TopicRegistry::send_existing_subscriber_eventfds: 发送eventfd={}失败", dup_eventfd);
                close(dup_eventfd);
            }
        }
        
        DEMO_INFO("TopicRegistry::send_existing_subscriber_eventfds: topic '{}' 成功发送 {}/{} 个EventFD给Publisher", 
                  topic_name, success_count, subscriber_count);
    }

    void accept_loop() {
        DEMO_INFO("TopicRegistry::accept_loop: 开始接受连接循环");
        while (running_) {
            struct sockaddr_un client_addr;
            socklen_t client_len = sizeof(client_addr);
            
            int client_socket = accept(server_socket_, (struct sockaddr*)&client_addr, &client_len);
            if (client_socket == -1) {
                if (running_) {
                    DEMO_WARN("TopicRegistry::accept_loop: accept失败, errno={}, 继续等待", errno);
                    continue;
                }
                DEMO_DEBUG("TopicRegistry::accept_loop: 服务停止，退出accept循环");
                break;
            }
            
            DEMO_DEBUG("TopicRegistry::accept_loop: 接受新连接, client_socket={}", client_socket);
            
            // 在新线程中处理客户端请求
            std::thread([this, client_socket]() {
                handle_client(client_socket);
            }).detach();
        }
        DEMO_INFO("TopicRegistry::accept_loop: 接受连接循环结束");
    }
    
    void handle_client(int client_socket) {
        DEMO_DEBUG("TopicRegistry::handle_client: 处理客户端连接, socket={}", client_socket);
        
        ClientRequest request;
        ssize_t bytes_read = recv(client_socket, &request, sizeof(request), 0);
        
        if (bytes_read != sizeof(request)) {
            DEMO_ERROR("TopicRegistry::handle_client: 接收请求失败, expected={}, actual={}, errno={}", 
                       sizeof(request), bytes_read, errno);
            close(client_socket);
            return;
        }
        
        std::string topic_name(request.topic_name);
        const char* type_str = (request.type == RequestType::SUBSCRIBE_CONNECT) ? "SUBSCRIBE" : "PUBLISH";
        DEMO_INFO("TopicRegistry::handle_client: 收到{}请求, topic='{}', socket={}", 
                  type_str, topic_name, client_socket);
        
        auto topic_info = get_or_create_topic(topic_name);
        if (!topic_info) {
            DEMO_ERROR("TopicRegistry::handle_client: 创建topic '{}'失败", topic_name);
            close(client_socket);
            return;
        }
        
        if (request.type == RequestType::SUBSCRIBE_CONNECT) {
            // 为订阅者创建专用的EventFD
            int subscriber_eventfd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
            if (subscriber_eventfd == -1) {
                DEMO_ERROR("TopicRegistry::handle_client: 创建eventfd失败, errno={}", errno);
                close(client_socket);
                return;
            }
            
            DEMO_DEBUG("TopicRegistry::handle_client: 为订阅者创建eventfd={}, topic='{}'", 
                       subscriber_eventfd, topic_name);
            
            topic_info->add_subscriber_eventfd(subscriber_eventfd);
            
            // 发送EventFD给订阅者
            if (EventFdHelper::send_fd(client_socket, subscriber_eventfd) > 0) {
                const char* ack = "OK";
                if (send(client_socket, ack, 2, 0) == 2) {
                    DEMO_INFO("TopicRegistry::handle_client: 成功处理订阅请求, topic='{}', eventfd={}", 
                              topic_name, subscriber_eventfd);
                    
                    // 注意：简化设计 - Publisher会在连接时获取已存在的EventFD
                    // 对于后来加入的订阅者，依赖Publisher定期重连或其他机制
                    DEMO_DEBUG("TopicRegistry::handle_client: 新订阅者已添加，Publisher需要重连获取更新的EventFD列表");
                } else {
                    DEMO_ERROR("TopicRegistry::handle_client: 发送确认失败, errno={}", errno);
                    topic_info->remove_subscriber_eventfd(subscriber_eventfd);
                }
            } else {
                DEMO_ERROR("TopicRegistry::handle_client: 发送eventfd失败");
                topic_info->remove_subscriber_eventfd(subscriber_eventfd);
            }
        } else if (request.type == RequestType::PUBLISH_CONNECT) {
            // 处理发布者连接：发送确认并传输已存在的订阅者EventFD列表
            const char* ack = "OK";
            if (send(client_socket, ack, 2, 0) == 2) {
                DEMO_INFO("TopicRegistry::handle_client: 成功发送发布确认, topic='{}'", topic_name);
                
                // 发送已存在的订阅者EventFD列表
                send_existing_subscriber_eventfds(client_socket, topic_info, topic_name);
                
                DEMO_INFO("TopicRegistry::handle_client: 成功处理发布请求, topic='{}'", topic_name);
            } else {
                DEMO_ERROR("TopicRegistry::handle_client: 发送确认失败, errno={}", errno);
            }
        } else {
            DEMO_ERROR("TopicRegistry::handle_client: 未知请求类型: {}", (int)request.type);
        }
        
        close(client_socket);
        DEMO_DEBUG("TopicRegistry::handle_client: 关闭客户端连接, socket={}", client_socket);
    }
    
    void cleanup_topics() {
        std::lock_guard<std::mutex> lock(topics_mutex_);
        DEMO_INFO("TopicRegistry::cleanup_topics: 清理 {} 个topic", topics_.size());
        topics_.clear();
    }
};

// 改进的Publisher类
class Publisher {
private:
    Config config_;
    std::unordered_map<std::string, std::shared_ptr<TopicInfo>> topics_;
    std::unordered_map<std::string, std::set<int>> topic_subscriber_eventfds_;  // 保存每个topic的订阅者EventFD列表
    std::mutex subscriber_eventfds_mutex_;  // 保护EventFD列表的访问
    
public:
    explicit Publisher(const Config& config = Config{}) : config_(config) {}
    
    ~Publisher() {
        cleanup();
    }
    
    bool connect_to_topic(const std::string& topic_name) {
        DEMO_INFO("Publisher::connect_to_topic: 连接到topic '{}'", topic_name);
        
        // 连接到topic注册服务
        int sock = socket(AF_UNIX, SOCK_STREAM, 0);
        if (sock == -1) {
            DEMO_ERROR("Publisher::connect_to_topic: 创建socket失败, errno={}", errno);
            return false;
        }
        DEMO_DEBUG("Publisher::connect_to_topic: 创建socket成功, fd={}", sock);
        
        struct sockaddr_un addr;
        memset(&addr, 0, sizeof(addr));
        addr.sun_family = AF_UNIX;
        strncpy(addr.sun_path, config_.registry_socket_path.c_str(), 
                sizeof(addr.sun_path) - 1);
        
        if (connect(sock, (struct sockaddr*)&addr, sizeof(addr)) == -1) {
            DEMO_ERROR("Publisher::connect_to_topic: 连接注册中心失败, path={}, errno={}", 
                       config_.registry_socket_path, errno);
            close(sock);
            return false;
        }
        DEMO_DEBUG("Publisher::connect_to_topic: 连接注册中心成功, path={}", config_.registry_socket_path);
        
        // 发送连接请求
        ClientRequest request;
        request.type = RequestType::PUBLISH_CONNECT;
        strncpy(request.topic_name, topic_name.c_str(), sizeof(request.topic_name) - 1);
        request.topic_name[sizeof(request.topic_name) - 1] = '\0';
        
        if (send(sock, &request, sizeof(request), 0) != sizeof(request)) {
            DEMO_ERROR("Publisher::connect_to_topic: 发送请求失败, errno={}", errno);
            close(sock);
            return false;
        }
        DEMO_DEBUG("Publisher::connect_to_topic: 发送PUBLISH请求成功, topic='{}'", topic_name);
        
        // 等待确认
        char ack[3];
        if (recv(sock, ack, 2, 0) != 2 || strncmp(ack, "OK", 2) != 0) {
            DEMO_ERROR("Publisher::connect_to_topic: 接收确认失败, topic='{}'", topic_name);
            close(sock);
            return false;
        }
        DEMO_DEBUG("Publisher::connect_to_topic: 收到注册中心确认");
        
        // 接收已存在的订阅者EventFD列表
        std::this_thread::sleep_for(std::chrono::milliseconds(1000));
        int count = receive_existing_subscriber_eventfds(sock, topic_name);
        DEMO_DEBUG("Publisher::connect_to_topic: 接收已存在的订阅者EventFD列表, 订阅者数={}", count);
        
        close(sock);
        
        // 直接连接到msgq
        auto topic_info = std::make_shared<TopicInfo>();
        std::string msgq_path = config_.msgq_prefix + topic_name;
        
        DEMO_DEBUG("Publisher::connect_to_topic: 连接msgq, path={}", msgq_path);
        if (msgq_new_queue(&topic_info->msgq, msgq_path.c_str(), config_.topic_buffer_size) < 0) {
            DEMO_ERROR("Publisher::connect_to_topic: 连接msgq失败, path={}", msgq_path);
            return false;
        }
        
        msgq_init_publisher(&topic_info->msgq);
        topic_info->topic_name = topic_name;
        topic_info->is_initialized = true;
        topics_[topic_name] = topic_info;
        
        DEMO_INFO("Publisher::connect_to_topic: 成功连接到topic '{}', msgq_path={}", topic_name, msgq_path);
        return true;
    }
    
    bool publish(const std::string& topic_name, const std::vector<uint8_t>& data) {
        DEMO_TRACE("Publisher::publish: 发布消息到topic '{}', 数据大小={}", topic_name, data.size());
        
        auto topic_it = topics_.find(topic_name);
        if (topic_it == topics_.end()) {
            // 自动连接到topic
            DEMO_DEBUG("Publisher::publish: topic '{}' 未连接，自动连接", topic_name);
            if (!connect_to_topic(topic_name)) {
                DEMO_ERROR("Publisher::publish: 自动连接topic '{}'失败", topic_name);
                return false;
            }
            topic_it = topics_.find(topic_name);
        }
        
        auto& topic_info = topic_it->second;
        
        // 使用msgq发送消息
        msgq_msg_t msg;
        if (msgq_msg_init_data(&msg, (char*)data.data(), data.size()) < 0) {
            DEMO_ERROR("Publisher::publish: msgq_msg_init_data失败, topic='{}', size={}", topic_name, data.size());
            return false;
        }
        
        int result = msgq_msg_send(&msg, &topic_info->msgq);
        msgq_msg_close(&msg);
        
        if (result > 0) {
            topic_info->message_count++;
            
            DEMO_DEBUG("Publisher::publish: 消息写入msgq成功, topic='{}', size={}, msg_count={}", 
                       topic_name, data.size(), topic_info->message_count.load());
            
            // 通知注册中心有新消息（触发所有订阅者的EventFD）
            notify_registry_new_message(topic_name);
            
            DEMO_INFO("Publisher::publish: 成功发布消息, topic='{}', size={}", topic_name, data.size());
            return true;
        } else {
            DEMO_ERROR("Publisher::publish: msgq_msg_send失败, topic='{}', result={}", topic_name, result);
        }
        
        return false;
    }
    
    // 添加新的订阅者EventFD到Publisher的本地列表
    void add_subscriber_eventfd(const std::string& topic_name, int eventfd) {
        std::lock_guard<std::mutex> lock(subscriber_eventfds_mutex_);
        topic_subscriber_eventfds_[topic_name].insert(eventfd);
        DEMO_INFO("Publisher::add_subscriber_eventfd: topic '{}' 添加订阅者eventfd={}, 总订阅者数={}", 
                  topic_name, eventfd, topic_subscriber_eventfds_[topic_name].size());
    }
    
    // 从Publisher的本地列表移除订阅者EventFD
    void remove_subscriber_eventfd(const std::string& topic_name, int eventfd) {
        std::lock_guard<std::mutex> lock(subscriber_eventfds_mutex_);
        auto topic_it = topic_subscriber_eventfds_.find(topic_name);
        if (topic_it != topic_subscriber_eventfds_.end()) {
            topic_it->second.erase(eventfd);
            DEMO_INFO("Publisher::remove_subscriber_eventfd: topic '{}' 移除订阅者eventfd={}, 剩余订阅者数={}", 
                      topic_name, eventfd, topic_it->second.size());
        }
    }
    
    // 从注册中心接收已存在的订阅者EventFD列表
    int receive_existing_subscriber_eventfds(int sock, const std::string& topic_name) {
        DEMO_DEBUG("Publisher::receive_existing_subscriber_eventfds: 接收topic '{}'的已存在订阅者列表", topic_name);
        
        // 接收订阅者数量
        uint32_t subscriber_count;
        if (recv(sock, &subscriber_count, sizeof(subscriber_count), 0) != sizeof(subscriber_count)) {
            DEMO_WARN("Publisher::receive_existing_subscriber_eventfds: 接收订阅者数量失败, topic='{}'", topic_name);
            return -1;
        }
        
        DEMO_DEBUG("Publisher::receive_existing_subscriber_eventfds: topic '{}' 有 {} 个已存在的订阅者", 
                   topic_name, subscriber_count);
        
        std::lock_guard<std::mutex> lock(subscriber_eventfds_mutex_);
        for (uint32_t i = 0; i < subscriber_count; i++) {
            int eventfd = EventFdHelper::recv_fd(sock);
            if (eventfd != -1) {
                topic_subscriber_eventfds_[topic_name].insert(eventfd);
                DEMO_DEBUG("Publisher::receive_existing_subscriber_eventfds: 接收到eventfd={} for topic '{}'", 
                           eventfd, topic_name);
            } else {
                DEMO_WARN("Publisher::receive_existing_subscriber_eventfds: 接收第{}个eventfd失败", i);
            }
        }
        
        DEMO_INFO("Publisher::receive_existing_subscriber_eventfds: topic '{}' 成功接收 {} 个订阅者EventFD", 
                  topic_name, topic_subscriber_eventfds_[topic_name].size());
        return topic_subscriber_eventfds_[topic_name].size();
    }

private:
    void notify_registry_new_message(const std::string& topic_name) {
        DEMO_DEBUG("Publisher::notify_registry_new_message: 直接通知topic '{}'的所有订阅者", topic_name);
        
        std::lock_guard<std::mutex> lock(subscriber_eventfds_mutex_);
        auto topic_it = topic_subscriber_eventfds_.find(topic_name);
        if (topic_it == topic_subscriber_eventfds_.end() || topic_it->second.empty()) {
            DEMO_DEBUG("Publisher::notify_registry_new_message: topic '{}' 暂无订阅者", topic_name);
            return;
        }
        
        uint64_t notification = 1;
        int success_count = 0;
        int fail_count = 0;
        
        DEMO_TRACE("Publisher::notify_registry_new_message: topic '{}' 通知 {} 个订阅者", 
                   topic_name, topic_it->second.size());
        
        for (int eventfd : topic_it->second) {
            if (write(eventfd, &notification, sizeof(notification)) == sizeof(notification)) {
                success_count++;
                DEMO_TRACE("Publisher::notify_registry_new_message: 成功通知eventfd={}", eventfd);
            } else {
                fail_count++;
                DEMO_WARN("Publisher::notify_registry_new_message: 通知eventfd={}失败, errno={}", eventfd, errno);
            }
        }
        
        DEMO_DEBUG("Publisher::notify_registry_new_message: topic '{}' 通知完成, 成功={}, 失败={}", 
                   topic_name, success_count, fail_count);
    }
    
    void cleanup() {
        DEMO_DEBUG("Publisher::cleanup: 清理 {} 个topic连接", topics_.size());
        for (auto& [topic, topic_info] : topics_) {
            if (topic_info->is_initialized) {
                DEMO_DEBUG("Publisher::cleanup: 关闭topic '{}' 的msgq连接", topic);
                msgq_close_queue(&topic_info->msgq);
            }
        }
        topics_.clear();
        
        // 清理EventFD（注意：不要关闭，这些EventFD由订阅者拥有）
        std::lock_guard<std::mutex> lock(subscriber_eventfds_mutex_);
        for (auto& [topic, eventfds] : topic_subscriber_eventfds_) {
            DEMO_DEBUG("Publisher::cleanup: 清理topic '{}' 的 {} 个订阅者EventFD引用", topic, eventfds.size());
        }
        topic_subscriber_eventfds_.clear();
        
        DEMO_INFO("Publisher::cleanup: Publisher清理完成");
    }
};

// 改进的Subscriber类
class Subscriber {
private:
    Config config_;
    boost::asio::io_context io_context_;
    std::unique_ptr<boost::asio::steady_timer> keep_alive_timer_;
    std::unordered_map<std::string, int> topic_eventfds_;
    std::unordered_map<std::string, std::shared_ptr<TopicInfo>> topics_;
    std::unordered_map<std::string, std::unique_ptr<boost::asio::posix::stream_descriptor>> descriptors_;
    std::unordered_map<std::string, MessageCallback> callbacks_;
    std::unique_ptr<std::thread> io_thread_;
    std::atomic<bool> running_;
    
public:
    explicit Subscriber(const Config& config = Config{}) 
        : config_(config), running_(false) {}
    
    ~Subscriber() {
        stop();
    }
    
    void start() {
        DEMO_INFO("Subscriber::start: 启动订阅者服务");
        running_ = true;
        
        // 创建长期定时器防止 io_context 提前退出
        keep_alive_timer_ = std::make_unique<boost::asio::steady_timer>(io_context_);
        keep_alive_timer_->expires_at(std::chrono::steady_clock::time_point::max());
        keep_alive_timer_->async_wait([](const boost::system::error_code&) {
            // 此回调永远不会被调用，只是保持 io_context 运行
        });
        
        io_thread_ = std::make_unique<std::thread>([this]() {
            DEMO_DEBUG("Subscriber::start: IO线程开始运行");
            io_context_.run();
            DEMO_DEBUG("Subscriber::start: IO线程结束运行");
        });
        
        // 等待一小段时间确保IO线程完全启动
        std::this_thread::sleep_for(std::chrono::milliseconds(50));
        DEMO_INFO("Subscriber::start: 订阅者服务启动完成");
    }
    
    void stop() {
        DEMO_INFO("Subscriber::stop: 停止订阅者服务");
        running_ = false;
        
        // 取消定时器并停止 io_context
        if (keep_alive_timer_) {
            keep_alive_timer_->cancel();
            keep_alive_timer_.reset();
        }
        io_context_.stop();
        
        if (io_thread_ && io_thread_->joinable()) {
            DEMO_DEBUG("Subscriber::stop: 等待IO线程结束");
            io_thread_->join();
        }
        
        cleanup();
        DEMO_INFO("Subscriber::stop: 订阅者服务已停止");
    }
    
    bool subscribe(const std::string& topic_name, MessageCallback callback) {
        DEMO_INFO("Subscriber::subscribe: 订阅topic '{}'", topic_name);
        
        // 连接到topic获取专用EventFD
        if (!connect_to_topic(topic_name)) {
            DEMO_ERROR("Subscriber::subscribe: 连接topic '{}'失败", topic_name);
            return false;
        }
        
        callbacks_[topic_name] = callback;
        DEMO_DEBUG("Subscriber::subscribe: 注册topic '{}'的回调函数", topic_name);
        
        // 创建异步EventFD监听
        auto eventfd = topic_eventfds_[topic_name];
        auto descriptor = std::make_unique<boost::asio::posix::stream_descriptor>(io_context_, eventfd);
        
        DEMO_DEBUG("Subscriber::subscribe: 开始异步监听topic '{}', eventfd={}", topic_name, eventfd);
        DEMO_DEBUG("Subscriber::subscribe: io_context线程状态: running={}", running_.load());
        
        start_async_read(topic_name, descriptor.get());
        descriptors_[topic_name] = std::move(descriptor);
        
        DEMO_INFO("Subscriber::subscribe: 成功订阅topic '{}'", topic_name);
        return true;
    }
    
private:
    bool connect_to_topic(const std::string& topic_name) {
        DEMO_INFO("Subscriber::connect_to_topic: 连接到topic '{}'", topic_name);
        
        int sock = socket(AF_UNIX, SOCK_STREAM, 0);
        if (sock == -1) {
            DEMO_ERROR("Subscriber::connect_to_topic: 创建socket失败, errno={}", errno);
            return false;
        }
        DEMO_DEBUG("Subscriber::connect_to_topic: 创建socket成功, fd={}", sock);
        
        struct sockaddr_un addr;
        memset(&addr, 0, sizeof(addr));
        addr.sun_family = AF_UNIX;
        strncpy(addr.sun_path, config_.registry_socket_path.c_str(), 
                sizeof(addr.sun_path) - 1);
        
        if (connect(sock, (struct sockaddr*)&addr, sizeof(addr)) == -1) {
            DEMO_ERROR("Subscriber::connect_to_topic: 连接注册中心失败, path={}, errno={}", 
                       config_.registry_socket_path, errno);
            close(sock);
            return false;
        }
        DEMO_DEBUG("Subscriber::connect_to_topic: 连接注册中心成功, path={}", config_.registry_socket_path);
        
        // 发送订阅请求
        ClientRequest request;
        request.type = RequestType::SUBSCRIBE_CONNECT;
        strncpy(request.topic_name, topic_name.c_str(), sizeof(request.topic_name) - 1);
        request.topic_name[sizeof(request.topic_name) - 1] = '\0';
        
        if (send(sock, &request, sizeof(request), 0) != sizeof(request)) {
            DEMO_ERROR("Subscriber::connect_to_topic: 发送订阅请求失败, errno={}", errno);
            close(sock);
            return false;
        }
        DEMO_DEBUG("Subscriber::connect_to_topic: 发送SUBSCRIBE请求成功, topic='{}'", topic_name);
        
        // 接收专用EventFD
        int eventfd = EventFdHelper::recv_fd(sock);
        if (eventfd == -1) {
            DEMO_ERROR("Subscriber::connect_to_topic: 接收eventfd失败, topic='{}'", topic_name);
            close(sock);
            return false;
        }
        DEMO_DEBUG("Subscriber::connect_to_topic: 接收专用eventfd成功, eventfd={}", eventfd);
        
        // 等待确认
        char ack[3];
        if (recv(sock, ack, 2, 0) != 2 || strncmp(ack, "OK", 2) != 0) {
            DEMO_ERROR("Subscriber::connect_to_topic: 接收确认失败, topic='{}'", topic_name);
            close(eventfd);
            close(sock);
            return false;
        }
        close(sock);
        DEMO_DEBUG("Subscriber::connect_to_topic: 收到注册中心确认");
        
        topic_eventfds_[topic_name] = eventfd;
        
        // 连接到msgq作为subscriber
        auto topic_info = std::make_shared<TopicInfo>();
        std::string msgq_path = config_.msgq_prefix + topic_name;
        
        DEMO_DEBUG("Subscriber::connect_to_topic: 连接msgq, path={}", msgq_path);
        if (msgq_new_queue(&topic_info->msgq, msgq_path.c_str(), config_.topic_buffer_size) < 0) {
            DEMO_ERROR("Subscriber::connect_to_topic: 连接msgq失败, path={}", msgq_path);
            close(eventfd);
            return false;
        }
        
        msgq_init_subscriber(&topic_info->msgq);
        topic_info->topic_name = topic_name;
        topic_info->is_initialized = true;
        topics_[topic_name] = topic_info;
        
        DEMO_INFO("Subscriber::connect_to_topic: 成功连接到topic '{}', eventfd={}, msgq_path={}", 
                  topic_name, eventfd, msgq_path);
        return true;
    }
    
    void start_async_read(const std::string& topic_name, boost::asio::posix::stream_descriptor* descriptor) {
        auto buffer = std::make_shared<uint64_t>();
        
        DEMO_TRACE("Subscriber::start_async_read: 开始异步读取topic '{}'", topic_name);
        
        descriptor->async_read_some(
            boost::asio::buffer(buffer.get(), sizeof(*buffer)),
            [this, topic_name, descriptor, buffer](const boost::system::error_code& ec, std::size_t bytes) {
                if (!running_) {
                    DEMO_DEBUG("Subscriber::start_async_read: 停止监听topic '{}'", topic_name);
                    return;
                }
                
                if (!ec && bytes == sizeof(*buffer)) {
                    DEMO_DEBUG("Subscriber::start_async_read: 收到EventFD通知, topic='{}', value={}", 
                               topic_name, *buffer);
                    
                    // 有新消息到达，从msgq读取
                    process_messages(topic_name);
                    
                    // 继续监听
                    start_async_read(topic_name, descriptor);
                } else {
                    DEMO_WARN("Subscriber::start_async_read: 异步读取失败, topic='{}', ec={}, bytes={}, 重新开始监听", 
                              topic_name, ec.message(), bytes);
                    
                    // 即使出错也要继续监听（可能是暂时性错误）
                    if (running_) {
                        std::this_thread::sleep_for(std::chrono::milliseconds(10));
                        start_async_read(topic_name, descriptor);
                    }
                }
            }
        );
    }
    
    void process_messages(const std::string& topic_name) {
        DEMO_TRACE("Subscriber::process_messages: 处理topic '{}'的消息", topic_name);
        
        auto topic_it = topics_.find(topic_name);
        auto callback_it = callbacks_.find(topic_name);
        
        if (topic_it == topics_.end() || callback_it == callbacks_.end()) {
            DEMO_ERROR("Subscriber::process_messages: 找不到topic '{}'的信息或回调", topic_name);
            return;
        }
        
        auto& topic_info = topic_it->second;
        auto& callback = callback_it->second;
        
        // 从msgq读取所有可用消息
        msgq_msg_t msg;
        int message_count = 0;
        
        while (msgq_msg_ready(&topic_info->msgq)) {
            int recv_result = msgq_msg_recv(&msg, &topic_info->msgq);
            if (recv_result > 0) {
                message_count++;
                std::vector<uint8_t> data(msg.data, msg.data + msg.size);
                
                DEMO_DEBUG("Subscriber::process_messages: 从msgq读取消息, topic='{}', size={}", 
                           topic_name, msg.size);
                
                Message message(topic_name, data);
                message.timestamp = std::chrono::duration_cast<std::chrono::milliseconds>(
                    std::chrono::system_clock::now().time_since_epoch()).count();
                
                try {
                    callback(message);
                    DEMO_TRACE("Subscriber::process_messages: 成功调用回调, topic='{}'", topic_name);
                } catch (const std::exception& e) {
                    DEMO_ERROR("Subscriber::process_messages: 回调函数异常, topic='{}', error={}", 
                               topic_name, e.what());
                }
                
                msgq_msg_close(&msg);
            } else {
                DEMO_DEBUG("Subscriber::process_messages: msgq_msg_recv返回{}, 停止读取", recv_result);
                break;
            }
        }
        
        if (message_count > 0) {
            DEMO_INFO("Subscriber::process_messages: 处理完成, topic='{}', 消息数={}", 
                      topic_name, message_count);
        } else {
            DEMO_DEBUG("Subscriber::process_messages: 没有可用消息, topic='{}'", topic_name);
        }
    }
    
    void cleanup() {
        DEMO_DEBUG("Subscriber::cleanup: 清理 {} 个topic连接", topics_.size());
        
        for (auto& [topic, topic_info] : topics_) {
            if (topic_info->is_initialized) {
                DEMO_DEBUG("Subscriber::cleanup: 关闭topic '{}' 的msgq连接", topic);
                msgq_close_queue(&topic_info->msgq);
            }
        }
        
        for (auto& [topic, eventfd] : topic_eventfds_) {
            DEMO_DEBUG("Subscriber::cleanup: 关闭topic '{}' 的eventfd {}", topic, eventfd);
            close(eventfd);
        }
        
        descriptors_.clear();
        topic_eventfds_.clear();
        topics_.clear();
        callbacks_.clear();
        
        DEMO_INFO("Subscriber::cleanup: Subscriber清理完成");
    }
};

} // namespace improved_pubsub
